package com.echo.im.offlinePush.mq.receiver;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.TypeReference;
import com.echo.im.common.model.mq.OfflineMsgMqInfo;
import com.echo.im.common.contant.MQConstants;
import com.echo.im.infrastructure.mq.event.BaseEvent;
import com.echo.im.offlinePush.service.PushOfflineMessageService;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 私聊消息存储队列消费者 - 使用手动ACK和批量拉取、批量插入数据库
 * @author echo
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class OfflineMessageReceiver {
    private final PushOfflineMessageService pushOfflineMessageService;
    // 批量处理的最大消息数
    private static final int BATCH_SIZE = 200;

    /**
     * 监听RabbitMQ队列中的私聊消息
     * - 手动ACK模式由`ackMode = "MANUAL"`指定。
     * - 批量预取值在注解中的`prefetch`配置，设置为10，表示一次最多拉取10条消息。
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = MQConstants.OFFLINE_MESSAGE_PUSH_QUEUE, durable = "true"),
                    exchange = @Exchange(value = MQConstants.ECHO_STORE_EXCHANGE, type = ExchangeTypes.TOPIC),
                    key = MQConstants.OFFLINE_MESSAGE_TOPIC
            ),
            ackMode = "MANUAL",  // 手动ACK模式
            containerFactory = "rabbitListenerContainerFactoryWithPrefetch"  // 配置批量预取值
    )
    public void onChatMessage(List<Message> messages, Channel channel) throws Exception {

        List<Long> deliveryTags = new ArrayList<>();
        List<OfflineMsgMqInfo> offlineMsgMqInfos = new ArrayList<>();  // 用于批量插入的列表

        try {
            // 遍历处理从队列中拉取到的每条消息
            for (Message message : messages) {
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);

                // 将消息体转换为自定义的业务对象
                BaseEvent.EventMessage<OfflineMsgMqInfo> eventMessage
                        = JSON.parseObject(msg, new TypeReference<BaseEvent.EventMessage<OfflineMsgMqInfo>>(){}.getType());

                OfflineMsgMqInfo data = eventMessage.getData();

                offlineMsgMqInfos.add(data);  // 添加到批量插入列表

                // 收集消息的deliveryTag以便批量确认
                deliveryTags.add(message.getMessageProperties().getDeliveryTag());

                // 如果已达到批量大小，立即进行数据库批量插入和批量ACK
                if (offlineMsgMqInfos.size() == BATCH_SIZE) {
                    doReceive(offlineMsgMqInfos, deliveryTags, channel);
                }
            }

            // 处理完所有消息后，如果还有未插入和未ACK的消息，进行最后的批量操作
            if (!offlineMsgMqInfos.isEmpty()) {
                doReceive(offlineMsgMqInfos, deliveryTags, channel);
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("处理消息出现异常：{}", e.getMessage());
            // 如果出现异常，拒绝处理并重新投递这些消息
            for (Long deliveryTag : deliveryTags) {
                channel.basicNack(deliveryTag, false, false); // 重新投递
            }
        }
    }


    private void doReceive(List<OfflineMsgMqInfo> offlineMsgMqInfos, List<Long> deliveryTags, Channel channel) throws Exception {
        try {
            for (OfflineMsgMqInfo offlineMsgMqInfo : offlineMsgMqInfos){
                pushOfflineMessageService.pushOfflineMessage(offlineMsgMqInfo);
            }

            // 批量确认消息，使用最后一个deliveryTag进行批量确认
            long lastDeliveryTag = deliveryTags.get(deliveryTags.size() - 1);
            channel.basicAck(lastDeliveryTag, true);  // true表示确认所有小于等于此deliveryTag的消息
            log.info("批量ACK确认 {} 条消息", deliveryTags.size());

            // 清空已插入和确认的消息列表
            offlineMsgMqInfos.clear();
            deliveryTags.clear();
        } catch (Exception e) {
            log.error("批量插入或ACK出现异常：{}", e.getMessage());
            throw e;
        }
    }
}
